package vip.xiaonuo.common.kafka;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class KafkaMessageProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public KafkaMessageProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 发送一条消息到指定的主题。
     *
     * @param topic 目标主题名称
     * @param key   消息键（可选）
     * @param value 消息值
     */
    public void sendMessage(String topic, String key, String value) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, value);
        // 添加回调来处理发送结果
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.printf("Message sent to the Kafka topic: %s, partition: %d, offset: %d%n",
                        metadata.topic(), metadata.partition(), metadata.offset());
            } else {
                System.err.println("Unable to send message due to : " + ex.getMessage());
            }
        });
    }

    /**
     * 同步发送一条消息到指定的主题。
     *
     * @param topic 目标主题名称
     * @param key   消息键（可选）
     * @param value 消息值
     * @throws Exception 如果发送失败抛出异常
     */
    public void sendSyncMessage(String topic, String key, String value) throws Exception {
        try {
            SendResult<String, String> result = kafkaTemplate.send(topic, key, value).get();
            RecordMetadata metadata = result.getRecordMetadata();
            System.out.printf("Message sent synchronously to the Kafka topic: %s, partition: %d, offset: %d%n",
                    metadata.topic(), metadata.partition(), metadata.offset());
        } catch (Exception e) {
            throw new Exception("Failed to send message", e);
        }
    }
}
